package org.zjt.rxjava.lession.second;

import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
import javafx.beans.value.ChangeListener;
import javafx.beans.value.ObservableValue;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

/**
 * Description:
 * <p>
 *   Disposable 订阅、发布的控制器
 *
 *      1、订阅控制器
 *      2、订阅Observer 中的控制器。
 *      3、订阅外部 interval 控制器
 *      4、发送者的控制器
 *
 *
 * @author juntao.zhang
 * Date: 2018-06-28 下午3:21
 * @see
 */
@Slf4j
public class DisposableTest {

    public static void main(String[] args) throws InterruptedException {

        /** 组合 控制器 **/
        CompositeDisposable compositeDisposable = new CompositeDisposable();

        Observable<Long> source2 = Observable.interval(1, TimeUnit.SECONDS).map(item -> item * 3);
        Disposable subscribe = source2.subscribe(item -> log.info(String.valueOf(item)));
        Disposable subscribe2 = source2.subscribe(item -> log.info(String.valueOf(item)));

        compositeDisposable.add(subscribe);
        compositeDisposable.add(subscribe2);

        compositeDisposable.dispose();

        TimeUnit.SECONDS.sleep(4);




        log.info("--------over--------");

    }


    private static <T> Observable<T> valuesOf(final ObservableValue<T> fxObservable) {

        return Observable.create(observableEmitter -> {

            //emit initial state
            observableEmitter.onNext(fxObservable.getValue());

            //emit value changes uses a listener
            final ChangeListener<T> listener = (observableValue, prev, current) -> observableEmitter.onNext(current);

            //add listener to ObservableValue
            fxObservable.addListener(listener);

            //Handle disposing by specifying cancellable
            observableEmitter.setCancellable(() -> fxObservable.removeListener(listener));
        });
    }

    /**
     * 在方法外部执行订阅可执行判断
     */
    public static void handlingOutObserver() {
        try {
            Observable<Long> source2 = Observable.interval(1, TimeUnit.SECONDS).map(item -> item * 3);
            Disposable subscribe = source2.subscribe(item -> log.info(String.valueOf(item)));
            subscribe.dispose();
            TimeUnit.SECONDS.sleep(4);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }


    /***
     * 在方法内执行 订阅可执行判断
     */
    public static void handlingInObserver() {
        Observable<String> observable = Observable.generate(emitter -> {
            emitter.onNext("zhang");
            emitter.onComplete();
        });

        Observer observer = new Observer() {

            /**
             * 执行订阅前判断是否可执行
             * @param d
             */
            @Override
            public void onSubscribe(Disposable d) {
                /***
                 *  不能提出。不能执行。不会执行下面的订阅方法。
                 */
                if (!d.isDisposed()) {
                    d.dispose();
                }
            }

            @Override
            public void onNext(Object o) {
                log.info(String.valueOf(o));
            }

            @Override
            public void onError(Throwable e) {
                log.error("", e);
            }

            @Override
            public void onComplete() {
                log.info("complete");
            }
        };

        observable.subscribe(observer);
    }


    /**
     * 发送者内部的响应订阅者Disposed，判断是否可发送
     */
    public static void observableDispose() {
        ObservableOnSubscribe observableOnSubscribe = emitter -> {
            for (int i = 0; i < 20; i++) {
                TimeUnit.SECONDS.sleep(1);
                if (!emitter.isDisposed()) {
                    emitter.onNext(i);
                }
            }
            emitter.onComplete();
        };

        Observable.create(observableOnSubscribe).subscribe(new Observer() {

            /**
             * 执行订阅前判断是否可执行
             * @param d
             */
            @Override
            public void onSubscribe(Disposable d) {
                /***
                 *  不能提出。不能执行。不会执行下面的订阅方法。
                 */
                if (!d.isDisposed()) {
                    d.dispose();
                }
            }

            @Override
            public void onNext(Object o) {
                log.info(String.valueOf(o));
            }

            @Override
            public void onError(Throwable e) {
                log.error("", e);
            }

            @Override
            public void onComplete() {
                log.info("complete");
            }
        });
    }
}
